package com.song.hadoop01;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SortMapperReduce  extends Configured implements Tool {
	
	
	private  static class SortMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable>
	{
		
		private static IntWritable data = new IntWritable();
		@Override
		protected void map(LongWritable key, Text value,
				Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context)
						throws IOException, InterruptedException {
			
			int  val =  Integer.parseInt(value.toString());
			data.set(val);
			context.write(data, new IntWritable(1));
			
			
		}

	}
	
	public static class SortReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>
	{
		
		private static IntWritable numline = new IntWritable(1);
		@Override
		protected void reduce(IntWritable key, Iterable<IntWritable> items,
				Reducer<IntWritable, IntWritable, IntWritable, IntWritable>.Context context)
						throws IOException, InterruptedException {
			for(IntWritable item :items)
			{
				context.write(numline, key);
			}
			
			numline.set(numline.get()+1);
			
		}
		
	}
	
	/**
	 *  假如有多个节点会出现问题
	 * @author root
	 *
	 */
	public static class Partition extends Partitioner<IntWritable, IntWritable>
	{

		@Override
		public int getPartition(IntWritable key, IntWritable value, int numPartitions) {
			int maxnumber =65223;
			
			int bound = maxnumber  /numPartitions +1;
			int keynum = key.get();
			
			for(int i = 0 ; i< numPartitions ;i++)
			{
				if(keynum  < bound *(i+1) && keynum >= bound * i)
					return i;
			}
			
			return -1;
		}
		
	}
	
	public static void main(String[] args) {
		
		try {
			ToolRunner.run(new SortMapperReduce() , args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}



	public int run(String[] args) throws Exception {
		Configuration  conf = new Configuration();
		Job job = new Job(conf , "sort");
		job.setJarByClass(SortMapperReduce.class);
		job.setMapperClass(SortMapper.class);
		job.setReducerClass(SortReduce.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);
		job.setPartitionerClass(Partition.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		System.exit(job.waitForCompletion(true)? 0:1 );
		return 0;
	}
}

